package org.activequant.data.util;

import java.util.Comparator;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicReference;

import org.activequant.core.domainmodel.InstrumentSpecification;
import org.activequant.core.domainmodel.SeriesSpecification;
import org.activequant.core.domainmodel.data.Candle;
import org.activequant.core.domainmodel.data.MarketDataEntity;
import org.activequant.core.domainmodel.data.Quote;
import org.activequant.core.domainmodel.data.TickerNews;
import org.activequant.core.domainmodel.data.TradeIndication;
import org.activequant.core.domainmodel.marketdepth.MarketDepthEvent;
import org.activequant.core.types.TimeFrame;
import org.activequant.core.types.TimeStamp;
import org.activequant.data.retrieval.ICandleSubscriptionSource;
import org.activequant.data.retrieval.IMarketDataReplayService;
import org.activequant.data.retrieval.IMarketDepthEventSubscriptionSource;
import org.activequant.data.retrieval.IQuoteSubscriptionSource;
import org.activequant.data.retrieval.ISeriesDataIteratorSource;
import org.activequant.data.retrieval.ISubscription;
import org.activequant.data.retrieval.ITickerNewsSubscriptionSource;
import org.activequant.data.retrieval.ITradeIndicationSubscriptionSource;
import org.activequant.math.algorithms.MergeSortIterator;
import org.apache.log4j.Logger;

/**
 * Aligns series of different market entities, and emits them as a single stream of
 * mixed entities: Quotes for different instruments, mixed with TradeIndications,
 * and Candles. The output sequence presents the order in which events occurred
 * (i.e. different data sources are correctly aligned).
 * <p>
 * In other words, the output is the timeline-based output. See {@link IMarketDataReplayService} 
 * for more details. <br/>
 * An IMarketDataReplayService implementation. Holds the following associated variables:
 * <ul>
 * <li>currentTimeStamp(AtomicReference&lt;TimeStamp&gt;)</li>
 * <li>comparator(Comparator&lt;MarketDataEntity&lt;?&gt;&gt;)</li>
 * <li>mergeSort(MergeSortIterator&lt;MarketDataEntity&lt;?&gt;&gt;)</li>
 * <li>quoteReplay(ReplayServiceBase&lt;Quote&gt;)</li>
 * <li>tradeReplay(ReplayServiceBase&lt;TradeIndication&gt;)</li>
 * <li>candleReplay(ReplayServiceBase&lt;Candle&gt;)</li>
 * <li>marketDepthEventReplay(ReplayServiceBase&lt;MarketDepthEvent&gt;)</li>
 * <li>tickerNewsReplay(ReplayServiceBase&lt;TickerNews&gt;)</li>
 * </ul>
 * <p>
 * <b>History:</b><br>
 *  - [26.11.2007] Created (Ulrich Staudinger)<br>
 *  - [27.11.2007] Generalized to support any market events (Mike Kroutikov)<br>
 *  - [13.01.2008] Major cleanup (Mike Kroutikov)<br>
 *  - [06.10.2008] In case of tie, Candles are fired last (Mike Kroutikov)<br>
 *
 *  @author Ulrich Staudinger
 *  @author Mike Kroutikov
 */
public class MarketDataReplayService implements IMarketDataReplayService {
	
	private final Logger log = Logger.getLogger(getClass());
	/**
	 * private final AtomicReference&lt;TimeStamp&gt; currentTimeStamp = new AtomicReference&lt;TimeStamp&gt;();
	 */
	private final AtomicReference<TimeStamp> currentTimeStamp = new AtomicReference<TimeStamp>();
	/**
	 * final Comparator&lt;MarketDataEntity&lt;?&gt;&gt; comparator = new Comparator&lt;MarketDataEntity&lt;?&gt;&gt;() {...}<br/>
	 * compares MarketDataEntity subclasses by their timeStamps. Candles go last (if same timeStamps show up).
	 */
	final Comparator<MarketDataEntity<?>> comparator = new Comparator<MarketDataEntity<?>>() {
		public int compare(MarketDataEntity<?> o1, MarketDataEntity<?> o2) {
			try{
			int rc = o1.getTimeStamp().compareTo(o2.getTimeStamp());
			if(rc != 0) return rc;
			
			// in case market data entities have the same timestamp, make sure that
			// Candles will come last. A candle may have same timestamp as TradeIndication that
			// was generated by this candle (this implies that candle series is used in backtesting
			// context as source of generated trade indications <em>and</em> as data feed).
			// Since in non-simulated context trade indications that are used to build a Candle
			// are always happen <em>before</em> the candle is fired, we better make simulated
			// data feed behave the same way.
			if(o1.getClass() == Candle.class || o2.getClass() == Candle.class) {
				if(o1.getClass() == Candle.class) {
					if(o2.getClass() == Candle.class) {
						return 0; // candle-to-candle
					} else {
						return 1;
					}
				} else {
					return -1;
				}
			}
			}catch(Exception e){}
			return 0;
		}
	};
	/**
	 * final MergeSortIterator&lt;MarketDataEntity&lt;?&gt;&gt; mergeSort = new MergeSortIterator&lt;MarketDataEntity&lt;?&gt;&gt;(comparator);<br/>
	 * a MergeSortIterator which sorts and merges data from feeds by their timeStamp
	 */
	final MergeSortIterator<MarketDataEntity<?>> mergeSort = new MergeSortIterator<MarketDataEntity<?>>(comparator);
	/**
	 * private final ReplayServiceBase&lt;Quote&gt; quoteReplay = new ReplayServiceBase&lt;Quote&gt;() {...}
	 */
	private final ReplayServiceBase<Quote> quoteReplay = new ReplayServiceBase<Quote>() {
		/**
		 * Creates and returns a SeriesSpecification using the instrumentSpecification(InstrumentSpecification) of the given entity(Quote) and TIMEFRAME_1_TICK to set its 
		 * associated instrumentSpecification(InstrumentSpecification) and timeFrame(TimeFrame)
		 */
		@Override
		protected SeriesSpecification inferSeriesSpecification(Quote entity) {
			return new SeriesSpecification(entity.getInstrumentSpecification(), TimeFrame.TIMEFRAME_1_TICK);
		}
		/**
		 * <strong>1.</strong> Gets a prot(Iterator&lt;Quote&gt;) to iterate the given iterator(Iterator&lt;Quote&gt;) in the interval starting after the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;)
		 * and ending with the endTimeStamp(TimeStamp) of the associated quoteReplay(ReplayServiceBase&lt;Quote&gt;)<br/>
		 * <strong>2.</strong> Adds a new Stream object to the streams(Queue&lt;Stream&gt;) of the associated mergeSort(MergeSortIterator&lt;MarketDataEntity&lt;?&gt;&gt;)
		 * using that prot(Iterator&lt;Quote&gt;)
		 */
		@SuppressWarnings("unchecked")
		@Override
		protected void processFetchedIterator(Iterator<Quote> iterator) {
			TimeStamp low = new TimeStamp(currentTimeStamp.get().getNanoseconds() + 1L);
			Iterator<Quote> prot = new ProtectTimeIntervalIterator<Quote>(iterator, low, quoteReplay.getEndTimeStamp());
			mergeSort.addIterator((Iterator<MarketDataEntity<?>>) (Iterator<?>) prot);
		}
	};
	/**
	 * private final ReplayServiceBase&lt;TradeIndication&gt; tradeReplay = new ReplayServiceBase&lt;TradeIndication&gt;() {...}
	 */
	private final ReplayServiceBase<TradeIndication> tradeReplay = new ReplayServiceBase<TradeIndication>() {
		/**
		 * Creates and returns a SeriesSpecification using the instrumentSpecification(InstrumentSpecification) of the given entity(TradeIndication) and TIMEFRAME_1_TICK to set its 
		 * associated instrumentSpecification(InstrumentSpecification) and timeFrame(TimeFrame)
		 */
		@Override
		protected SeriesSpecification inferSeriesSpecification(TradeIndication entity) {
			return new SeriesSpecification(entity.getInstrumentSpecification(), TimeFrame.TIMEFRAME_1_TICK);
		}
		/**
		 * <strong>1.</strong> Gets a prot(Iterator&lt;TradeIndication&gt;) to iterate the given iterator(Iterator&lt;TradeIndication&gt;) in the interval starting after the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;)
		 * and ending with the endTimeStamp(TimeStamp) of the associated tradeReplay(ReplayServiceBase&lt;TradeIndication&gt;)<br/>
		 * <strong>2.</strong> Adds a new Stream object to the streams(Queue&lt;Stream&gt;) of the associated mergeSort(MergeSortIterator&lt;MarketDataEntity&lt;?&gt;&gt;)
		 * using that prot(Iterator&lt;TradeIndication&gt;)
		 */
		@SuppressWarnings("unchecked")
		@Override
		protected void processFetchedIterator(Iterator<TradeIndication> iterator) {
			TimeStamp low = new TimeStamp(currentTimeStamp.get().getNanoseconds() + 1L);
			Iterator<TradeIndication> prot = new ProtectTimeIntervalIterator<TradeIndication>(iterator, low, tradeReplay.getEndTimeStamp());
			mergeSort.addIterator((Iterator<MarketDataEntity<?>>) (Iterator<?>) prot);
		}
	};
	/**
	 * private final ReplayServiceBase&lt;Candle&gt; candleReplay = new ReplayServiceBase&lt;Candle&gt;() {...}
	 */
	private final ReplayServiceBase<Candle> candleReplay = new ReplayServiceBase<Candle>() {
		/**
		 * Creates and returns a SeriesSpecification using the instrumentSpecification(InstrumentSpecification) and timeFrame(TimeFrame) of the given entity(Candle) to set its 
		 * associated instrumentSpecification(InstrumentSpecification) and timeFrame(TimeFrame)
		 */
		@Override
		protected SeriesSpecification inferSeriesSpecification(Candle entity) {
			return new SeriesSpecification(entity.getInstrumentSpecification(), entity.getTimeFrame());
		}
		/**
		 * <strong>1.</strong> Gets a prot(Iterator&lt;Candle&gt;) to iterate the given iterator(Iterator&lt;Candle&gt;) in the interval starting after the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;)
		 * and ending with the endTimeStamp(TimeStamp) of the associated candleReplay(ReplayServiceBase&lt;Candle&gt;)<br/>
		 * <strong>2.</strong> Adds a new Stream object to the streams(Queue&lt;Stream&gt;) of the associated mergeSort(MergeSortIterator&lt;MarketDataEntity&lt;?&gt;&gt;)
		 * using that prot(Iterator&lt;Candle&gt;)
		 */
		@SuppressWarnings("unchecked")
		@Override
		protected void processFetchedIterator(Iterator<Candle> iterator) {
			TimeStamp low = new TimeStamp(currentTimeStamp.get().getNanoseconds() + 1L);
			Iterator<Candle> prot = new ProtectTimeIntervalIterator<Candle>(iterator, low, candleReplay.getEndTimeStamp());
			mergeSort.addIterator((Iterator<MarketDataEntity<?>>) (Iterator<?>) prot);
		}
	};
	/**
	 * private final ReplayServiceBase&lt;MarketDepthEvent&gt; marketDepthEventReplay = new ReplayServiceBase&lt;MarketDepthEvent&gt;() {...}
	 */
	private final ReplayServiceBase<MarketDepthEvent> marketDepthEventReplay = new ReplayServiceBase<MarketDepthEvent>() {
		/**
		 * Creates and returns a SeriesSpecification using the instrumentSpecification(InstrumentSpecification) of the given entity(MarketDepthEvent) and TIMEFRAME_1_TICK to set its 
		 * associated instrumentSpecification(InstrumentSpecification) and timeFrame(TimeFrame)
		 */
		@Override
		protected SeriesSpecification inferSeriesSpecification(MarketDepthEvent entity) {
			return new SeriesSpecification(entity.getInstrumentSpecification(), TimeFrame.TIMEFRAME_1_TICK);
		}
		/**
		 * <strong>1.</strong> Gets a prot(Iterator&lt;MarketDepthEvent&gt;) to iterate the given iterator(Iterator&lt;MarketDepthEvent&gt;) in the interval starting after the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;)
		 * and ending with the endTimeStamp(TimeStamp) of the associated marketDepthEventReplay(ReplayServiceBase&lt;MarketDepthEvent&gt;)<br/>
		 * <strong>2.</strong> Adds a new Stream object to the streams(Queue&lt;Stream&gt;) of the associated mergeSort(MergeSortIterator&lt;MarketDataEntity&lt;?&gt;&gt;)
		 * using that prot(Iterator&lt;MarketDepthEvent&gt;)
		 */
		@SuppressWarnings("unchecked")
		@Override
		protected void processFetchedIterator(Iterator<MarketDepthEvent> iterator) {
			TimeStamp low = new TimeStamp(currentTimeStamp.get().getNanoseconds() + 1L);
			Iterator<MarketDepthEvent> prot = new ProtectTimeIntervalIterator<MarketDepthEvent>(iterator, low, marketDepthEventReplay.getEndTimeStamp());
			mergeSort.addIterator((Iterator<MarketDataEntity<?>>) (Iterator<?>) prot);
		}
	};
	/**
	 * private final ReplayServiceBase&lt;TickerNews&gt; tickerNewsReplay = new ReplayServiceBase&lt;TickerNews&gt;() {...}
	 */
	private final ReplayServiceBase<TickerNews> tickerNewsReplay = new ReplayServiceBase<TickerNews>() {
		/**
		 * Creates and returns a SeriesSpecification using the instrumentSpecification(InstrumentSpecification) of the given entity(TickerNews) and TIMEFRAME_1_TICK to set its 
		 * associated instrumentSpecification(InstrumentSpecification) and timeFrame(TimeFrame)
		 */
		@Override
		protected SeriesSpecification inferSeriesSpecification(TickerNews entity) {
			return new SeriesSpecification(entity.getInstrumentSpecification(), TimeFrame.TIMEFRAME_1_TICK);
		}
		/**
		 * <strong>1.</strong> Gets a prot(Iterator&lt;TickerNews&gt;) to iterate the given iterator(Iterator&lt;TickerNews&gt;) in the interval starting after the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;)
		 * and ending with the endTimeStamp(TimeStamp) of the associated tickerNewsReplay(ReplayServiceBase&lt;TickerNews&gt;)<br/>
		 * <strong>2.</strong> Adds a new Stream object to the streams(Queue&lt;Stream&gt;) of the associated mergeSort(MergeSortIterator&lt;TickerNews&lt;?&gt;&gt;)
		 * using that prot(Iterator&lt;TickerNews&gt;)
		 */
		@SuppressWarnings("unchecked")
		@Override
		protected void processFetchedIterator(Iterator<TickerNews> iterator) {
			TimeStamp low = new TimeStamp(currentTimeStamp.get().getNanoseconds() + 1L);
			Iterator<TickerNews> prot = new ProtectTimeIntervalIterator<TickerNews>(iterator, low, tickerNewsReplay.getEndTimeStamp());
			mergeSort.addIterator((Iterator<MarketDataEntity<?>>) (Iterator<?>) prot);
		}
	};
	
	/**
	 * Start of the historical interval.<br/>
	 * returns the startTimeStamp(TimeStamp) of the associated quoteReplay(ReplayServiceBase&lt;Quote&gt;)
	 * @return start.
	 */
	public TimeStamp getStartTimeStamp() {
		return quoteReplay.getStartTimeStamp();
	}
	/**
	 * <strong>1.</strong> Sets the startTimeStamp(TimeStamp) and value of the currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) of the associated quoteReplay(ReplayServiceBase&lt;Quote&gt;) with the given val(TimeStamp)<br/>
	 * <strong>2.</strong> Sets the startTimeStamp(TimeStamp) and value of the currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) of the associated tradeReplay(ReplayServiceBase&lt;TradeIndication&gt;) with the given val(TimeStamp)<br/>
	 * <strong>3.</strong> Sets the startTimeStamp(TimeStamp) and value of the currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) of the associated marketDepthEventReplay(ReplayServiceBase&lt;MarketDepthEvent&gt;) with the given val(TimeStamp)<br/>
	 * <strong>4.</strong> Sets the startTimeStamp(TimeStamp) and value of the currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) of the associated candleReplay(ReplayServiceBase&lt;Candle&gt;) with the given val(TimeStamp)<br/>
	 * <strong>5.</strong> Sets the startTimeStamp(TimeStamp) and value of the currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) of the associated tickerNewsReplay(ReplayServiceBase&lt;TickerNews&gt;) with the given val(TimeStamp)<br/>
	 * <strong>5.</strong> Sets the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) with the given val(TimeStamp)
	 * @param val
	 */
	public void setStartTimeStamp(TimeStamp val) {
		quoteReplay.setStartTimeStamp(val);
		tradeReplay.setStartTimeStamp(val);
		candleReplay.setStartTimeStamp(val);
		marketDepthEventReplay.setStartTimeStamp(val);
		tickerNewsReplay.setStartTimeStamp(val);
		currentTimeStamp.set(val);
	}
	
	/**
	 * End of the historical interval.<br/>
	 * returns the endTimeStamp(TimeStamp) of the associated quoteReplay(ReplayServiceBase&lt;Quote&gt;)
	 * @return end.
	 */
	public TimeStamp getEndTimeStamp() {
		return quoteReplay.getEndTimeStamp();
	}
	/**
	 * <strong>1.</strong> Sets the endTimeStamp(TimeStamp) of the associated quoteReplay(ReplayServiceBase&lt;Quote&gt;) with the given val(TimeStamp)<br/>
	 * <strong>2.</strong> Sets the endTimeStamp(TimeStamp) of the associated tradeReplay(ReplayServiceBase&lt;TradeIndication&gt;) with the given val(TimeStamp)<br/>
	 * <strong>3.</strong> Sets the endTimeStamp(TimeStamp) of the associated candleReplay(ReplayServiceBase&lt;Candle&gt;) with the given val(TimeStamp)<br/>
	 * <strong>4.</strong> Sets the endTimeStamp(TimeStamp) of the associated marketDepthEventReplay(ReplayServiceBase&lt;MarketDepthEvent&gt;) with the given val(TimeStamp)<br/>
	 * <strong>5.</strong> Sets the endTimeStamp(TimeStamp) of the associated tickerNewsReplay(ReplayServiceBase&lt;TickerNews&gt;) with the given val(TimeStamp)
	 * @param val
	 */
	public void setEndTimeStamp(TimeStamp val) {
		quoteReplay.setEndTimeStamp(val);
		tradeReplay.setEndTimeStamp(val);
		candleReplay.setEndTimeStamp(val);
		marketDepthEventReplay.setEndTimeStamp(val);
		tickerNewsReplay.setEndTimeStamp(val);
	}
	
	/**
	 * Historic data source for quotes.<br/>
	 * returns the seriesSource(ISeriesDataIteratorSource&lt;Quote&gt;) of the associated quoteReplay(ReplayServiceBase&lt;Quote&gt;)
	 * @return data source.
	 */
	public ISeriesDataIteratorSource<Quote> getQuoteSeriesDataSource() {
		return quoteReplay.getSeriesDataSource();
	}
	
	/**
	 * Sets historic data source for quotes.<br/>
	 * sets the seriesSource(ISeriesDataIteratorSource&lt;Quote&gt;) of the associated quoteReplay(ReplayServiceBase&lt;Quote&gt;) with the given val(ISeriesDataIteratorSource&lt;Quote&gt;)
	 * @param val data source.
	 */
	public void setQuoteSeriesDataSource(ISeriesDataIteratorSource<Quote> val) {
		quoteReplay.setSeriesDataSource(val);
	}
	
	/**
	 * Historic data source for trade indications.<br/>
	 * returns the seriesSource(ISeriesDataIteratorSource&lt;TradeIndication&gt;) of the associated tradeReplay(ReplayServiceBase&lt;TradeIndication&gt;)
	 * @return data source.
	 */
	public ISeriesDataIteratorSource<TradeIndication> getTradeIndicationSeriesDataSource() {
		return tradeReplay.getSeriesDataSource();
	}
	
	/**
	 * Sets historic data source for trade indications.<br/>
	 * sets the seriesSource(ISeriesDataIteratorSource&lt;TradeIndication&gt;) of the associated tradeReplay(ReplayServiceBase&lt;TradeIndication&gt;) with the given val(ISeriesDataIteratorSource&lt;TradeIndication&gt;)
	 * @param val data source.
	 */
	public void setTradeIndicationSeriesDataSource(ISeriesDataIteratorSource<TradeIndication> val) {
		tradeReplay.setSeriesDataSource(val);
	}
	/**
	 * Historic data source for candles.<br/>
	 * returns the seriesSource(ISeriesDataIteratorSource&lt;Candle&gt;) of the associated candleReplay(ReplayServiceBase&lt;Candle&gt;)
	 * @return data source.
	 */
	public ISeriesDataIteratorSource<Candle> getCandleSeriesDataSource() {
		return candleReplay.getSeriesDataSource();
	}
	/**
	 * Sets historic data source for candles.<br/>
	 * sets the seriesSource(ISeriesDataIteratorSource&lt;Candle&gt;) of the associated candleReplay(ReplayServiceBase&lt;Candle&gt;) with the given val(ISeriesDataIteratorSource&lt;Candle&gt;)
	 * @param val data source.
	 */
	public void setCandleSeriesDataSource(ISeriesDataIteratorSource<Candle> val) {
		candleReplay.setSeriesDataSource(val);
	}
	/**
	 * Historic data source for marketDepth events.<br/>
	 * returns the seriesSource(ISeriesDataIteratorSource&lt;MarketDepthEvent&gt;) of the associated marketDepthEventReplay(ReplayServiceBase&lt;MarketDepthEvent&gt;)
	 * @return data source.
	 */
	public ISeriesDataIteratorSource<MarketDepthEvent> getMarketDepthEventSeriesDataSource() {
		return marketDepthEventReplay.getSeriesDataSource();
	}	
	/**
	 * Sets historic data source for marketDepth events.<br/>
	 * sets the seriesSource(ISeriesDataIteratorSource&lt;MarketDepthEvent&gt;) of the associated marketDepthEventReplay(ReplayServiceBase&lt;MarketDepthEvent&gt;) with the given val(ISeriesDataIteratorSource&lt;TickerNews&gt;)
	 * @param val data source.
	 */
	public void setMarketDepthEventSeriesDataSource(ISeriesDataIteratorSource<MarketDepthEvent> val) {
		marketDepthEventReplay.setSeriesDataSource(val);
	}
	/**
	 * Historic data source for tickerNews events.<br/>
	 * returns the seriesSource(ISeriesDataIteratorSource&lt;TickerNews&gt;) of the associated tickerNewsReplay(ReplayServiceBase&lt;TickerNews&gt;)
	 * @return data source.
	 */
	public ISeriesDataIteratorSource<TickerNews> getTickerNewsSeriesDataSource() {
		return tickerNewsReplay.getSeriesDataSource();
	}
	
	/**
	 * Sets historic data source for tickerNews events.<br/>
	 * sets the seriesSource(ISeriesDataIteratorSource&lt;TickerNews&gt;) of the associated tickerNewsReplay(ReplayServiceBase&lt;TickerNews&gt;) with the given val(ISeriesDataIteratorSource&lt;TickerNews&gt;)
	 * @param val data source.
	 */
	public void setTickerNewsSeriesDataSource(ISeriesDataIteratorSource<TickerNews> val) {
		tickerNewsReplay.setSeriesDataSource(val);
	}
	/**
	 * <strong>1.</strong> Sets the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) with the startTimeStamp(TimeStamp)
	 * of the associated quoteReplay(ReplayServiceBase&lt;Quote&gt;)<br/>
	 * <strong>2.</strong> Creates and returns an Iterator&lt;MarketDataEntity&lt;?&gt;&gt; which gets his next element from the associated mergeSort(MergeSortIterator&lt;MarketDataEntity&lt;?&gt;&gt;)
	 * and sets the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;) with the timeStamp of that next MarketDataEntity element
	 */
	public Iterator<MarketDataEntity<?>> iterator() {
		currentTimeStamp.set(quoteReplay.getStartTimeStamp());
		
		return new Iterator<MarketDataEntity<?>>() {
			
			public boolean hasNext() {
				return mergeSort.hasNext();
			}

			public MarketDataEntity<?> next() {
				MarketDataEntity<?> entity = mergeSort.next();
				currentTimeStamp.set(entity.getTimeStamp());
				return entity;
			}

			public void remove() {
				mergeSort.remove();
			}
		};
	}
	/**
	 * returns a new IQuoteSubscriptionSource which uses the associated quoteReplay(ReplayServiceBase&lt;Quote&gt;)
	 */
	public IQuoteSubscriptionSource getQuoteSubscriptionSource() {
		return new IQuoteSubscriptionSource() {
			/**
			 * returns an ISubscription&lt;Quote&gt;[] array from the associated quoteReplay(ReplayServiceBase&lt;Quote&gt;)
			 */
			public ISubscription<Quote>[] getSubscriptions() {
				return quoteReplay.getSubscriptions();
			}
			/**
			 * creates and returns an ISubscription&lt;Quote&gt; for the given spec(InstrumentSpecification) and TIMEFRAME_1_TICK
			 * using the associated quoteReplay(ReplayServiceBase&lt;Quote&gt;)
			 */
			public ISubscription<Quote> subscribe(InstrumentSpecification spec)
					throws Exception {
				return quoteReplay.subscribe(spec, TimeFrame.TIMEFRAME_1_TICK);
			}
		};
	}
	/**
	 * returns a new ITradeIndicationSubscriptionSource which uses the associated tradeReplay(ReplayServiceBase&lt;TradeIndication&gt;)
	 */
	public ITradeIndicationSubscriptionSource getTradeIndicationSubscriptionSource() {
		return new ITradeIndicationSubscriptionSource() {
			/**
			 * returns an ISubscription&lt;TradeIndication&gt;[] array from the associated tradeReplay(ReplayServiceBase&lt;TradeIndication&gt;)
			 */
			public ISubscription<TradeIndication>[] getSubscriptions() {
				return tradeReplay.getSubscriptions();
			}
			/**
			 * creates and returns an ISubscription&lt;TradeIndication&gt; for the given spec(InstrumentSpecification) and TIMEFRAME_1_TICK
			 * using the associated tradeReplay(ReplayServiceBase&lt;TradeIndication&gt;)
			 */
			public ISubscription<TradeIndication> subscribe(InstrumentSpecification spec)
					throws Exception {
				return tradeReplay.subscribe(spec, TimeFrame.TIMEFRAME_1_TICK);
			}
		};
	}
	/**
	 * returns a new ICandleSubscriptionSource which uses the associated candleReplay(ReplayServiceBase&lt;Candle&gt;)
	 */
	public ICandleSubscriptionSource getCandleSubscriptionSource() {
		return new ICandleSubscriptionSource() {
			/**
			 * returns an ISubscription&lt;Candle&gt;[] array from the associated candleReplay(ReplayServiceBase&lt;Candle&gt;)
			 */
			public ISubscription<Candle>[] getSubscriptions() {
				return candleReplay.getSubscriptions();
			}
			/**
			 * creates and returns an ISubscription&lt;Candle&gt; for the given spec(InstrumentSpecification) and timeFrame(TimeFrame)
			 * using the associated candleReplay(ReplayServiceBase&lt;Candle&gt;)
			 */
			public ISubscription<Candle> subscribe(InstrumentSpecification spec, TimeFrame timeFrame)
					throws Exception {
				return candleReplay.subscribe(spec, timeFrame);
			}
		};
	}
	/**
	 * returns a new IMarketDepthEventSubscriptionSource which uses the associated marketDepthEventReplay(ReplayServiceBase&lt;MarketDepthEvent&gt;)
	 */
	public IMarketDepthEventSubscriptionSource getMarketDepthEventSubscriptionSource() {
		return new IMarketDepthEventSubscriptionSource() {
			/**
			 * returns an ISubscription&lt;MarketDepthEvent&gt;[] array from the associated marketDepthEventReplay(ReplayServiceBase&lt;MarketDepthEvent&gt;)
			 */
			public ISubscription<MarketDepthEvent>[] getSubscriptions() {
				return marketDepthEventReplay.getSubscriptions();
			}
			/**
			 * creates and returns an ISubscription&lt;MarketDataEntity&gt; for the given spec(InstrumentSpecification) and TIMEFRAME_1_TICK
			 * using the associated marketDepthEventReplay(ReplayServiceBase&lt;MarketDepthEvent&gt;)
			 */
			public ISubscription<MarketDepthEvent> subscribe(InstrumentSpecification spec)
					throws Exception {
				return marketDepthEventReplay.subscribe(spec, TimeFrame.TIMEFRAME_1_TICK);
			}
		};
	}
	/**
	 * returns a new ITickerNewsSubscriptionSource which uses the associated tickerNewsReplay(ReplayServiceBase&lt;TickerNews&gt;)
	 */
	public ITickerNewsSubscriptionSource getTickerNewsSubscriptionSource() {
		return new ITickerNewsSubscriptionSource() {
			/**
			 * returns an ISubscription&lt;TickerNews&gt;[] array from the associated tickerNewsReplay(ReplayServiceBase&lt;TickerNews&gt;)
			 */
			public ISubscription<TickerNews>[] getSubscriptions() {
				return tickerNewsReplay.getSubscriptions();
			}
			/**
			 * creates and returns an ISubscription&lt;TickerNews&gt; for the given spec(InstrumentSpecification) and TIMEFRAME_1_TICK
			 * using the associated tickerNewsReplay(ReplayServiceBase&lt;TickerNews&gt;)
			 */
			public ISubscription<TickerNews> subscribe(InstrumentSpecification spec)
					throws Exception {
				return tickerNewsReplay.subscribe(spec, TimeFrame.TIMEFRAME_1_TICK);
			}
		};
	}
	/**
	 * Depending on whether the given e(MarketDataEntity&lt;?&gt;) is a Quote, TradeIndication or Candle class/subclass it invokes
	 * the <code>quoteReplay.fire(e)</code>, <code>tradeReplay.fire(e)</code> or <code>candleReplay.fire(e)</code> method on one of the associated
	 * quoteReplay(ReplayServiceBase&lt;Quote&gt;), tradeReplay(ReplayServiceBase&lt;TradeIndication&gt;), candleReplay(ReplayServiceBase&lt;Candle&gt;), marketDepthEventReplay(ReplayServiceBase&lt;MarketDepthEvent&gt;) or tickerNewsReplay(ReplayServiceBase&lt;TickerNews&gt;)
	 */
	public void fire(MarketDataEntity<?> e) {
		log.debug("[dispatcher] fire: " + e);
				
		if(Quote.class.isAssignableFrom(e.getClass())) {
			quoteReplay.fire((Quote) (Object) e);
		} else if(TradeIndication.class.isAssignableFrom(e.getClass())) {
			tradeReplay.fire((TradeIndication) (Object) e);
		} else if(Candle.class.isAssignableFrom(e.getClass())) {
			candleReplay.fire((Candle) (Object) e);
		} else if(MarketDepthEvent.class.isAssignableFrom(e.getClass())) {
			marketDepthEventReplay.fire((MarketDepthEvent) (Object) e);
		} else if(TickerNews.class.isAssignableFrom(e.getClass())) {
			tickerNewsReplay.fire((TickerNews) (Object) e);
		}
	}
	/**
	 * returns the value of the associated currentTimeStamp(AtomicReference&lt;TimeStamp&gt;)
	 */
	public TimeStamp currentTimeStamp() {
		return currentTimeStamp.get();
	}
	/**
	 * returns the size of the streams(Queue&lt;Stream&gt;) of the associated mergeSort(MergeSortIterator&lt;MarketDataEntity&lt;?&gt;&gt;)
	 * @return
	 */
	public int getStreamsSize(){
		return mergeSort.getStreamsSize();
	}
}